package avicit.bdp.dcs.json.service.util;

import avicit.bdp.common.dto.api.APIHttpDefinition;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.desensitization.dto.DataMaskingMethodDto;
import avicit.bdp.dcs.desensitization.dto.DataMaskingPolicyDto;
import avicit.bdp.dcs.desensitization.dto.TransformerDto;
import avicit.bdp.dcs.desensitization.service.DataMaskingMethodService;
import avicit.bdp.dcs.desensitization.service.DataMaskingPolicyService;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.plugin.BaseDataxParam;
import avicit.bdp.dcs.plugin.DataxJsonUtil;
import avicit.bdp.dcs.plugin.TransformerParam;
import avicit.bdp.dcs.plugin.api.param.ApiDataxParam;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.SpeedLimitDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.HiveDataType;
import avicit.bdp.dcs.utils.DataTypeConstants;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.bdp.dcs.utils.JdbcConstants;
import avicit.platform6.core.exception.BusinessException;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

@Component
public class BuildJsonUtil {

  private static final Logger logger = LoggerFactory.getLogger(BuildJsonUtil.class);

  @Autowired private DataMaskingMethodService dataMaskingMethodService;
  @Autowired private DataMaskingPolicyService dataMaskingPolicyService;
  @Autowired private DataSourceUtils dataSourceUtils;

  private static final String DST_COLUMN_NAME = "dstColumnName";

  /** 列名转换 */
  public String convertColumnName(String dbType, String columnName) {
    switch (dbType) {
      case JdbcConstants.POSTGRESQL:
      case JdbcConstants.GREENPLUM:
        // 解决GP字段为关键字,insert出错
        return "\"" + columnName.toLowerCase() + "\"";
      case JdbcConstants.MYSQL:
        return "`" + columnName.toLowerCase() + "`";
      case JdbcConstants.DM:
      case JdbcConstants.ORACLE:
        return "\"" + columnName.toUpperCase() + "\"";
      case JdbcConstants.HIVE:
      case JdbcConstants.IOTDB:
      case JdbcConstants.ODPS:
        return columnName.toLowerCase();
      default:
        logger.warn(DcsErrorCode.UNSUPPORTED_DB_TYPE.getDesc(), dbType);
        return columnName;
    }
  }

  /** 转换数据类型 */
  public String convertDataType(String dbType, String dataType) {
    String hiveDataType = null;

    if (JdbcConstants.MYSQL.equals(dbType)) {

      switch (dataType.toLowerCase()) {
        case DataTypeConstants.TINYINT:
          hiveDataType = HiveDataType.TINYINT.getDescp();
          break;
        case DataTypeConstants.SMALLINT:
          hiveDataType = HiveDataType.SMALLINT.getDescp();
          break;
        case DataTypeConstants.MEDIUMINT:
        case DataTypeConstants.INT:
          hiveDataType = HiveDataType.INT.getDescp();
          break;
        case DataTypeConstants.BIGINT:
          hiveDataType = HiveDataType.BIGINT.getDescp();
          break;
        case DataTypeConstants.FLOAT:
          hiveDataType = HiveDataType.FLOAT.getDescp();
          break;
        case DataTypeConstants.DOUBLE:
        case DataTypeConstants.DECIMAL:
          hiveDataType = HiveDataType.DOUBLE.getDescp();
          break;
        case DataTypeConstants.DATETIME:
        case DataTypeConstants.TIMESTAMP:
        case DataTypeConstants.TIME:
          hiveDataType = HiveDataType.TIMESTAMP.getDescp();
          break;
        case DataTypeConstants.DATE:
          hiveDataType = HiveDataType.DATE.getDescp();
          break;
        default:
          hiveDataType = HiveDataType.STRING.getDescp();
          break;
      }

    } else if (JdbcConstants.POSTGRESQL.equals(dbType) || JdbcConstants.GREENPLUM.equals(dbType)) {
      if (dataType.contains(DataTypeConstants.INTEGER)) {
        hiveDataType = HiveDataType.INT.getDescp();
      } else if (dataType.contains(DataTypeConstants.DOUBLE_PRECISION)) {
        hiveDataType = HiveDataType.DOUBLE.getDescp();
      } else if (dataType.contains(DataTypeConstants.NUMERIC)) {
        hiveDataType = HiveDataType.DOUBLE.getDescp();
      } else if (dataType.contains(DataTypeConstants.TIMESTAMP)) {
        hiveDataType = HiveDataType.TIMESTAMP.getDescp();
      } else if (dataType.contains(DataTypeConstants.DATE)) {
        hiveDataType = HiveDataType.DATE.getDescp();
      } else if (dataType.contains(DataTypeConstants.REAL)) {
        hiveDataType = HiveDataType.DOUBLE.getDescp();
      } else {
        hiveDataType = HiveDataType.STRING.getDescp();
      }
    } else if (JdbcConstants.ORACLE.equals(dbType) || JdbcConstants.DM.equals(dbType)) {

      switch (dataType.toLowerCase()) {
        case DataTypeConstants.TINYINT:
          hiveDataType = HiveDataType.TINYINT.getDescp();
          break;
        case DataTypeConstants.SMALLINT:
        case DataTypeConstants.BYTE:
          hiveDataType = HiveDataType.SMALLINT.getDescp();
          break;
        case DataTypeConstants.INTEGER:
        case DataTypeConstants.INT:
          hiveDataType = HiveDataType.INT.getDescp();
          break;
        case DataTypeConstants.BIGINT:
          hiveDataType = HiveDataType.BIGINT.getDescp();
          break;
        case DataTypeConstants.FLOAT:
          hiveDataType = HiveDataType.FLOAT.getDescp();
          break;
        case DataTypeConstants.DOUBLE:
        case DataTypeConstants.DOUBLE_PRECISION:
          hiveDataType = HiveDataType.DOUBLE.getDescp();
          break;
        case DataTypeConstants.NUMERIC:
        case DataTypeConstants.NUMBER:
        case DataTypeConstants.DECIMAL:
        case DataTypeConstants.DEC:
          hiveDataType = HiveDataType.DECIMAL.getDescp();
          break;
        case DataTypeConstants.DATETIME:
        case DataTypeConstants.TIMESTAMP:
        case DataTypeConstants.TIME:
        case DataTypeConstants.TIME_WITH_TIME_ZONE:
        case DataTypeConstants.TIMESTAMP_WITH_TIME_ZONE:
        case DataTypeConstants.TIMESTAMP_WITH_LOCAL_TIME_ZONE:
        case DataTypeConstants.DATETIME_WITH_TIME_ZONE:
          hiveDataType = HiveDataType.TIMESTAMP.getDescp();
          break;
        case DataTypeConstants.DATE:
          hiveDataType = HiveDataType.DATE.getDescp();
          break;
        default:
          hiveDataType = HiveDataType.STRING.getDescp();
          break;
      }
    } else {
      logger.warn(DcsErrorCode.UNSUPPORTED_DB_TYPE.getDesc(), dbType);
    }
    return hiveDataType;
  }

  public List<String> getWriterColumns(TaskDTO taskDTO, String dbType) {
    List<String> writerColumns = new ArrayList<>();
    JSONArray jsonArray = JSON.parseArray(taskDTO.getMappingColumn());
    for (int i = 0; i < jsonArray.size(); i++) {
      JSONObject jsonObject = jsonArray.getJSONObject(i);
      writerColumns.add(convertColumnName(dbType, jsonObject.getString(DST_COLUMN_NAME)));
    }

    return writerColumns;
  }

  /** 解析脱敏参数 */
  public List<TransformerParam> buildTransformer(TaskDTO taskDTO) {
    List<TransformerParam> transformerParamList = new ArrayList<>();
    List<ColumnMappingDto> columnMappingDtoList =
        JSONUtils.toList(taskDTO.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        if (columnMappingDto.getTransformer() != null) {
          TransformerDto transformerDto = columnMappingDto.getTransformer();
          if (StringUtils.isEmpty(transformerDto.getId())
              || transformerDto.getColumnIndex() == null) {
            throw new BusinessException("脱敏方法配置有误，未配置id或列下标");
          }

          transformerParamList.add(buildTransformerParam(transformerDto));
        }
      }
    }

    return transformerParamList;
  }

  /** TransformerDto转变为TransformerParam */
  public TransformerParam buildTransformerParam(TransformerDto transformerDto) {
    TransformerParam param = new TransformerParam(null, new TransformerParam.Parameter());

    // step1:查询脱敏策略、脱敏算法
    DataMaskingPolicyDto dataMaskingPolicyDto =
        this.dataMaskingPolicyService.queryDataMaskingPolicyById(transformerDto.getId());
    DataMaskingMethodDto dataMaskingMethodDto =
        this.dataMaskingMethodService.queryDataMaskingDto(dataMaskingPolicyDto.getAlgoId());

    // step2:设置列下标
    param.getParameter().setColumnIndex(transformerDto.getColumnIndex());

    // step3:设置输入参数，如果有inner_type参数，需要设置内部参数
    List<String> inParas = Lists.newArrayList();
    String paras = dataMaskingPolicyDto.getParas();
    if (StringUtils.isNotEmpty(paras)) {
      inParas.addAll(Arrays.asList(dataMaskingPolicyDto.getParas().trim().split(Constants.COMMA)));
    }
    if (dataMaskingMethodDto.getInnerType() != null) {
      inParas.add(0, dataMaskingMethodDto.getInnerType());
    }
    param.getParameter().setParas(inParas);
    param.setName(dataMaskingMethodDto.getFuncName());

    return param;
  }

  public String buildJobJson(
      String srcDatasourceId,
      String dstDatasourceId,
      BaseDataxParam readerParam,
      BaseDataxParam writerParam,
      List<TransformerParam> transformerParamList,
      JobDTO jobDTO,
      TaskDTO taskDTO)
      throws Exception {

    DataxJsonUtil dataxJsonUtil = new DataxJsonUtil();

    String speedLimitParam = taskDTO.getSpeedLimitParam();
    if (StringUtils.isNotEmpty(speedLimitParam)) {
      SpeedLimitDto speedLimitDto = JSONObject.parseObject(speedLimitParam, SpeedLimitDto.class);
      dataxJsonUtil.setRecordLimit(speedLimitDto.getRecordLimit());
      dataxJsonUtil.setByteLimit(speedLimitDto.getByteLimit());
    }

    try {
      Map<String, Object> extraParam =
          JSONObject.parseObject(
              jobDTO.getExtraParams(), new TypeReference<Map<String, Object>>() {});
      readerParam.setExtraParam(extraParam);
      writerParam.setExtraParam(extraParam);
    } catch (Exception e) {
      logger.error("解析json error " + e.getMessage());
    }

    // 1、初始化reader
    if (jobDTO.getJobType().equals(JobType.API_HTTP.getCode())) {
      // 初始化errorLimitRecord
      ApiDataxParam dataxParam = (ApiDataxParam) readerParam;
      APIHttpDefinition apiHttpDefinition = dataxParam.getApiHttpDefinition();
      dataxJsonUtil.setErrorLimitRecord(apiHttpDefinition.getErrorLimitRecord());

      dataxJsonUtil.initReader(Constants.DB_TYPE_API, readerParam);
    } else {
      String systemBuilt = jobDTO.getExtraParams();
      JobDatasource readerDatasource =
          dataSourceUtils.convertDatasource(srcDatasourceId, systemBuilt);

      readerParam.setJdbcUrl(readerDatasource.getJdbcUrl());
      readerParam.setJdbcUsername(readerDatasource.getJdbcUsername());
      readerParam.setEncryptJdbcPassword(readerDatasource.getEncryptJdbcPassword());
      dataxJsonUtil.initReader(readerDatasource.getDbType(), readerParam);
    }

    // 2、初始化writer
    JobDatasource writerDatasource = dataSourceUtils.convertDatasource(dstDatasourceId);
    writerParam.setJdbcUrl(writerDatasource.getJdbcUrl());
    writerParam.setJdbcUsername(writerDatasource.getJdbcUsername());
    writerParam.setEncryptJdbcPassword(writerDatasource.getEncryptJdbcPassword());
    dataxJsonUtil.initWriter(writerDatasource.getDbType(), writerParam);

    // 3、初始化transformer
    dataxJsonUtil.initTransformer(transformerParamList);

    return JSON.toJSONString(dataxJsonUtil.buildJob());
  }
}
